package org.zhadoop.stream.api.streams;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;
import java.util.Locale;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.zhadoop.stream.cql.CQLUtils;

import com.thoughtworks.xstream.annotations.XStreamAlias;
import com.thoughtworks.xstream.annotations.XStreamAsAttribute;
import com.thoughtworks.xstream.annotations.XStreamImplicit;
import com.thoughtworks.xstream.annotations.XStreamOmitField;

/**
 * 流的Schema信息
 * 不论什么流，都必须包含一个schema
 * schema包含了列名称和列类型
 * 这些Schema一定是按照顺序来的
 * 
 * 本来在底层有IEventType 作为Schmea的
 * 但是由于属性过多，会导致序列化的时候很多冗余的信息，所以这里就建立了一个新的，
 * 待后面提交任务时候再进行转换
 * @author zwd
 *
 */
public class Schema implements Serializable{

	/**
	 * 序列化
	 */
	private static final long serialVersionUID = 3153151405297742211L;
	
	private static final Logger LOG = LoggerFactory.getLogger(Schema.class);
	
	/**
	 * Schema id.
	 * 在CQL解析阶段，会自动将streamName转为id
	 */
	@XStreamAlias("id")
	@XStreamAsAttribute
	private String id;
	
	/**
	 * Schema 名称
	 * 就是Schema的别名，这个别名是允许重复的
	 * 但是在CQL中，同一条CQL语句中，别名不能重复
	 * 在IDE中，名称就是schema的别名
	 * 在CQL中，name就是From子句中的别名
	 */
	@XStreamAlias("name")
	@XStreamAsAttribute
	private String name;
	
	/**
	 * 流名称
	 * 这个流名称在执行器中使用，
	 * 在XML API中，a.id这个a既可以是schema的名称
	 * 也可以是流名称
	 */
	@XStreamOmitField
	private String streamName;
	
	/**
	 * 列信息
	 */
	@XStreamImplicit(itemFieldName = "attribute")
	private List<Column> cols = new ArrayList<Column>();
	
	/**
	 * 构造函数
	 */
	public Schema(String id) {
		super();
		this.id = id.toLowerCase(Locale.US);
		this.name = id;
	}

	/**
	 * 根据名称检查该列是否已经添加到了schema中
	 * @param schemaName
	 * @return
	 */
	public boolean isAttributeExist(String schemaName){
		if(null == cols){
			return false;
		}
		for(Column attr : cols){
			if(attr.getName().equals(schemaName)){
				return true;
			}
		}
		return false;
	}
	
	
	
	/**
	 * 添加列信息到schema中
	 * @param attribute
	 */
	public void addCol(Column attribute){
		cols.add(attribute);
	}
	
	
	public Schema  cloneSchema(){
		Schema s = new Schema(id);
		s.setName(this.name);
		for(int j = 0 ; j < cols.size() ; j++){
			String colTyple = cols.get(j).getType();
			String colName = cols.get(j).getName();
			Class<?> type = getColumnTypeClass(colTyple);
			if(type == null){
				return null;
			}
			s.addCol(new Column(colName, type));
		}
		return s;
	}
	
	
	private Class<?> getColumnTypeClass(String colTyple) {
		if(colTyple == null){
			return null;
		}
		try{
			return Class.forName(colTyple, true, CQLUtils.getClassLoader());
		}catch(ClassNotFoundException e){
			LOG.error("failed to get attribute type {}", colTyple, e);
            return null;
		}
	}

	public String getId() {
		return id;
	}

	public void setId(String id) {
		this.id = id;
	}

	public String getName() {
		return name;
	}

	public void setName(String name) {
		this.name = name;
	}

	public String getStreamName() {
		return streamName;
	}

	public void setStreamName(String streamName) {
		this.streamName = streamName;
	}

	public List<Column> getCols() {
		return cols;
	}

	public void setCols(List<Column> cols) {
		this.cols = cols;
	}
}
